package syR

import (
	"database/sql"
	"db2s/Er"
	metaD "db2s/MetaInit"
	"db2s/arg"
	cp "db2s/componentInit"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/limit"
	"db2s/outPut"
	"db2s/parDef"
	"db2s/pubM"
	quit "db2s/quitDispos"
	mq "db2s/topic-mq"
	"errors"
	"fmt"
	"github.com/panjf2000/ants/v2"
	"os"
	"reflect"
	"sync"
	"time"
)

const (
	singleQueryMessageQueueLength int = 1
)

var (
	recoverCost = func(s string, r any) error {
		return errors.New(fmt.Sprintf("%v defer recover An exception was captured. abnormal is %v ", s, r))
	}
	recoverException = func(event string) {
		if re := recover(); re != nil {
			log.ErrorLog().Error(fmt.Sprintf("%v", recoverCost(event, re)))
		}
		return
	}
)

// 全局初始化结构体
type (
	InitDSnsPoolResult struct {
		SDB, DDB                                           map[string]*sql.DB
		SMS, SMScurry, DMS, DMScurry                       map[string]int64
		SrcModifyParameterRecode, DstModifyParameterRecode []global.ModifyParameterResultS
	}
	ReadWriteFilePointS struct {
		ParameterRecord *os.File
		BreakPointFin   *os.File
		ActiveDataFin   *os.File
		TtyBarFin       *os.File
		ObjectResult    *os.File
	}
	TableObjectMiss struct {
		ObjectLogSeq int64
		ObjectName   string
	}
	FatherSon struct {
		FatherTags    map[string]any
		FatherSonTags map[string][]string
	}
	SendMsgInfo struct {
		CheckMod  CheckMode
		Topic     Topic
		TableName string
	}
	ExecFuncS struct {
		CallFunc  sendCallFuncAction
		Parameter any
		Delta     int64
	}
	chanSendMsg struct {
		Object            string
		SendData          chan any
		SendMsgInfo       SendMsgInfo
		SendMsgExecFunc   ExecFuncS
		FinishMsgExecFunc ExecFuncS
		SpeedLimitSwitch  bool
		SpeedLimit        *limit.RateLimiter
		Status            *int64
		SendMsg           sendTopicF
		FatherSon         FatherSon
	}
)

var (
	continueTableList     = make(map[string]int)
	ChanSchedulePlanTable chan any
	baseMsgSendBody       *mq.MonitorMsgMq
	meta                  *CheckSchemaMetaData
)

type (
	schemaInitContext SchemaInitContext
	// Topic 自定义Topic 类型 限制范围 避免出错
	Topic string
	Logo  string //dataFix Type 标识
	// CheckObject 自定义CheckObject 类型 限制范围 避免出错
	CheckObject string
	CheckMode   string // CheckMode
	IndexType   string //Index type distinction
	Keywords    string //data split keyword
)

const (
	syncIsSegmentQuery       Topic = "sycIsSegmentQuery"
	syncMissSegmentQuery     Topic = "syncMissSegmentQuery"
	syncIsBaseQuery          Topic = "sycIsBaseQuery"
	syncIndexConcurrentQuery Topic = "syncIndexConcurrentQuery"
	syncQueryTableSql        Topic = "syncQueryTableSql"
	syncQueryTableData       Topic = "syncQueryTableData"
	syncGeneralInsert        Topic = "syncGeneralInsert"
	syncWriteInsert          Topic = "syncWriteInsert"
)

var (
	lock           sync.Mutex
	SchemaContext  *schemaInitContext
	PlanContext    *SchedulePlanContext
	GlobalPConfigs *GlobalParametersConfig
	closeObject    quit.Close
	//producer 生产者
	producer mq.Producer
	//consumer 消费者
	consumer                mq.Consumer
	FuncRef                 map[string]reflect.Value
	ScheduleTaskInitFuncRef map[string]reflect.Value
	MqListenerSInitFunc     map[string]reflect.Value
	WLog                    *log.Logger
)

type (
	SchemaInitContextConfig struct {
		TableList     []string            //表对象清单
		GtCheckConfig arg.ConfigParameter //客户端传参
	}
	// GlobalISObject 全局表结构或结构初始化对象
	GlobalISObject struct {
		TableMetaData map[string]Er.TablesMetaInfoEr
	}

	GlobalParametersConfig struct { // SchedulePlanConfig 输入参数重新复制
		FilePoint ReadWriteFilePointS
		repair    arg.RepairS
		result    arg.Result
		rules     arg.RulesS
		dSns      arg.DSNsS
		structs   arg.StructS
		db        pubM.Conn
	}
	CheckSchemaMetaData struct {
		TableMD               []*metaD.TableMeta
		TableObject           []*arg.InputTableAttributes
		TableObjectFlow       chan any
		ObjectSum             int
		IgnoreObjectSum       int64
		schemaMetaMsgSendBody *mq.MonitorMsgMq
	}
	SchedulePlanContextConfig struct {
		TableList     []string
		BeginTime     time.Time
		GtCheckConfig arg.ConfigParameter
		MQConfig      *cp.MQConfig
	}
	Object struct {
		TaskSeq         int    //表校验序号
		ObjectLogSeq    int64  //表对象的日志号
		TableMiss       bool   //表是否存在
		Schema, Table   string //check schema Name and check table Name
		MSchema, MTable string //map check schema name and check table name
		RowsLimit       int    //限制表的查询/迁移行数
		TableJoin       string
		Role            string
		PartName        []string
	}
	ReadOptimizer struct {
		IndexColumn      []string                                        //check table index column
		IndexColumnSeq   map[string] /*column seq + column name*/ string //check table index column Seq
		IndexColumnType  string                                          //check table index type
		ChunkStartSource string                                          //schedule task type  value:src,dst,all
		IndexName        string                                          //index name
		IndexCardinality string                                          //index  Cardinality
		ChunkSum         int64
		Thread           *int64
	}
	// SchedulePlan2 执行计划操作对象
	SchedulePlan2 struct {
		Object            Object
		SrcParameter      parDef.Parameter
		DstParameter      parDef.Parameter
		baseMe            tabBaseMe
		MQ                *mq.MonitorMsgMq
		ReadOptimizer     global.ReadOptimizer
		ReadOptimizerNew  ReadOptimizer
		repair            arg.RepairS
		rules             arg.RulesS
		writeOptimizer    global.WriteOptimizer
		TableLocNumber    uint64
		Status            global.TableStatus
		subTaskInfo       *mq.SubTasks
		TPod              *outPut.TPod
		XlsTerminal       *outPut.WorkSheetTerminal
		XlsResult         *outPut.WorkSheetResult
		XlsMiss           *outPut.WorkSheetMiss
		SpeedLimitMonitor *mq.SpeedLimitMonitor
		sync              global.SyncPlan
		TableAttributes   arg.TableSubOptions
	}

	// SchedulePlanContext 执行计划上下文
	SchedulePlanContext struct {
		//多任务执行计划
		ChanSchedulePlanTable chan any //[]map[string] /*SchemaToTable*/ *SchedulePlan2
		//存储执行计划任务
		SchedulePlanTable any //map[string] /*SchemaToTable*/ *SchedulePlan2
		// 上下文任务线程池
		Pool *ants.Pool
		//自动关闭mq的组件
		AutoCloseMqComponentTable map[cp.CheckMode]*cp.AutoCloseMqComponent
		//进度条
		TaskBarTable *mq.TaskBar
		Rw           sync.Mutex
		BeginTime    time.Time
		wg           sync.WaitGroup
		TableMiss    chan TableObjectMiss
	}
	tabBaseMe struct {
		Col   any `json:"struct"`
		Index any `json:"index"`
		Part  any `json:"part"`
	}
	// SchemaInitContext 执行计划上下文
	SchemaInitContext struct {
		Seq               int
		Schema            string
		Table             string
		MSchema           string
		MTable            string
		TableMiss         bool
		ObjectLogSeq      int64
		TColumnMeta       global.TableMetaGather
		IndexData         metaD.IndexChooseResult
		Role              string
		TableSum          int
		IgnoreObjectSum   int64
		TableFree         *int64
		TableAccount      *int64
		BreakPointSum     *int64
		TableAttributes   arg.TableSubOptions
		Meta              global.TableMetaS
		BaseMe            tabBaseMe
		Options           parDef.Options
		ChanSchemaContext chan any //chan *SchemaInitContext
		TPod              *outPut.TPod
	}
	AutoCloseMqComponent struct {
		TopicStatistics  map[Topic]*int64
		UnsubscribeTopic []Topic
		CheckMode        CheckMode
		EndMark          bool
	}
)
